package com.yugao.fintech.draper.message.queue.support.broadcast;

import cn.hutool.core.util.TypeUtil;
import com.hazelcast.topic.Message;
import com.hazelcast.topic.MessageListener;
import com.yugao.fintech.draper.message.queue.client.HazelcastMQTemplate;
import com.yugao.fintech.draper.message.queue.interceptor.HazelcastMessageInterceptor;
import com.yugao.fintech.draper.message.queue.message.AbstractMessage;
import com.yugao.fintech.draper.message.queue.support.HazelcastMessageListener;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;

import java.util.List;

@Getter
@Setter
public abstract class AbstractHazelcastBroadcastMessageListener<T extends HazelcastBroadcastMessage> implements MessageListener<T>, HazelcastMessageListener<T> {
    /**
     * 消息类型
     */
    private final Class<T> messageType;

    /**
     * 主题topic
     */
    private final String topic;

    private HazelcastMQTemplate mqTemplate;

    @SneakyThrows
    protected AbstractHazelcastBroadcastMessageListener() {
        this.messageType = (Class<T>) TypeUtil.getTypeArgument(getClass(), 0);
        this.topic = messageType.newInstance().getTopic();
    }

    @Override
    public void onMessage(Message<T> message) {
        T messageObject = message.getMessageObject();

        try {
            consumeMessageBefore(messageObject);
            // 消费消息
            this.onMessage(messageObject);
        } finally {
            consumeMessageAfter(messageObject);
        }
    }

    protected void consumeMessageAfter(AbstractMessage message) {
        List<HazelcastMessageInterceptor> interceptors = mqTemplate.getInterceptors();
        interceptors.forEach(interceptor -> interceptor.consumeMessageBefore(message));
    }

    protected void consumeMessageBefore(AbstractMessage message) {
        List<HazelcastMessageInterceptor> interceptors = mqTemplate.getInterceptors();
        for (int i = interceptors.size() - 1; i >= 0; i--) {
            interceptors.get(i).consumeMessageAfter(message);
        }
    }

    @Override
    public void process(T message) {

    }
}
